package org.acghub.mtdb.core;

import org.acghub.mtdb.common.exception.DeadLockError;
import org.acghub.mtdb.common.exception.MtdbError;
import org.acghub.mtdb.util.StringUtil;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicLong;

/**
 * @author zhouhe
 */
public final class Transaction {
    private static final ThreadLocal<Transaction> TRANSACTION = new ThreadLocal<>();
    private final static AtomicLong totalCount = new AtomicLong();
    private final static AtomicLong totalFalse = new AtomicLong();
    private final static AtomicLong totalException = new AtomicLong();
    final Map<LogKey, Object> wrappers = new HashMap<>();
    private final List<SavePoint> logs = new ArrayList<>();
    private final Map<Lockey, LockeyHolder> lockeys = new HashMap<>();
    private final List<Runnable> lastCommitActions = new ArrayList<>();
    private final Map<String, Map<Object, Object>> cachedTRecord = new HashMap<>();

    static Transaction current() {
        return TRANSACTION.get();
    }

    static Transaction create() {
        Transaction self = TRANSACTION.get();
        if (self == null) {
            TRANSACTION.set(self = new Transaction());
        }
        return self;
    }

    static void rollback(int savepoint) {
        current().rollback0(savepoint);
    }

    public static SavePoint currentSavepoint() {
        List<SavePoint> points = current().getLogs();
        return points.get(points.size() - 1);
    }

    public static boolean isActive() {
        return current() != null;
    }

    static void destroy() {
        TRANSACTION.set(null);
    }

    static int savepoint() {
        return current().savepoint0();
    }

    List<SavePoint> getLogs() {
        return logs;
    }

    Lockey get(Lockey lockey) {
        return lockeys.get(lockey).lockey;
    }

    void addLastCommitActions(Runnable r) {
        lastCommitActions.add(r);
    }

    void perform(AbstractProcedure<?> p) throws Throwable {
        try {
            totalCount.incrementAndGet();
            Tables.rFlushLock();
            try {
                if (p.call0()) {
                    commitLogs(p);
                    runActions();
                } else {
                    totalFalse.incrementAndGet();
                    lastRollback();
                }
            } catch (Throwable e) {
                lastRollback();
                throw e;
            } finally {
                finish();
                Tables.rFlushUnlock();
            }
        } catch (Throwable e) {
            p.setCause(e);
            p.setSuccess(false);
            totalException.incrementAndGet();
            Trace.fatal("Transaction Perform Exception", e);
            throw e;
        }
    }

    private void runActions() {
        lastCommitActions.forEach(Runnable::run);
    }

    private void lastRollback() {
        try {
            for (int index = logs.size() - 1; index >= 0; --index) {
                logs.get(index).rollback();
            }
            logs.clear();
        } catch (Throwable err) {
            Trace.fatal("last rollback ", err);
            Runtime.getRuntime().halt(-1);
        }
    }

    private int realCommit(String pname) {
        try {
            int count = 0;
            for (SavePoint sp : logs) {
                count += sp.commit();
            }
            logs.clear();
            return count;
        } catch (Throwable t) {
            Trace.warn("transaction.commit " + StringUtil.quote(pname) + " throw exception, rollback...", t);
            throw t;
        }
    }

    private void commitLogs(AbstractProcedure<?> p) {
        if (realCommit(p.name()) == 0) {
            return;
        }
        try {
            int maxNestNotify = 255;
            for (int nest = 0; nest < maxNestNotify; ++nest) {
                if (realCommit(p.name()) == 0) {
                    return;
                }
            }
            Trace.fatal("reach maxNestNotify. proc=" + p.getClass().getName());
        } catch (Throwable e) {
            Trace.fatal("logNotify", e);
        }
        lastRollback();
    }

    private int savepoint0() {
        logs.add(new SavePoint());
        return logs.size();
    }

    void rollback0(int savepoint) {
        if (savepoint < 1 || savepoint > logs.size()) {
            throw new MtdbError("zdb: invalid savepoint " + savepoint + "@" + logs.size());
        }
        while (logs.size() >= savepoint) {
            logs.remove(logs.size() - 1).rollback();
        }
    }

    private void finish() {
        lastCommitActions.clear();
        wrappers.clear();
        lockeys.forEach((k, v) -> v.cleanup());
        lockeys.clear();
        cachedTRecord.clear();
    }

    <K, V> TRecord<K, V> getCachedTRecord(String tableName, K k) {
        Map<Object, Object> table = cachedTRecord.get(tableName);
        return table != null ? (TRecord<K, V>) table.get(k) : null;
    }

    <K, V> void addCachedTRecord(String tableName, TRecord<K, V> r) {
        cachedTRecord.computeIfAbsent(tableName, v -> new HashMap<>()).put(r.getKey(), r);
    }

    <K, V> void removeCachedTRecord(String tableName, K k) {
        Map<Object, Object> table = cachedTRecord.get(tableName);
        if (table != null) {
            table.remove(k);
        }
    }

    void rAddLockey(Lockey lockey) {
        if (lockeys.containsKey(lockey)) {
            return;
        }
        lockey.rLock();
        lockeys.put(lockey, new LockeyHolder(lockey, LockeyHolderType.READ));
    }

    void wAddLockey(Lockey lockey) {
        LockeyHolder holder = lockeys.get(lockey);
        if (holder == null) {
            lockey.wLock();
            lockeys.put(lockey, new LockeyHolder(lockey, LockeyHolderType.WRITE));
        } else if (holder.type == LockeyHolderType.READ) {
            holder.lockey.rUnlock();
            try {
                holder.lockey.wLock();
            } catch (DeadLockError e) {
                lockeys.remove(lockey);
                throw e;
            }
            holder.type = LockeyHolderType.WRITE;
        }
    }

    LockeyHolderType getLockeyHolderType(Lockey lockey) {
        LockeyHolder holder = lockeys.get(lockey);
        return holder == null ? LockeyHolderType.NONE : holder.type;
    }

    enum LockeyHolderType {
        WRITE, READ, NONE
    }

    private static class LockeyHolder implements Comparable<LockeyHolder> {
        final Lockey lockey;
        LockeyHolderType type;

        LockeyHolder(Lockey lockey, LockeyHolderType type) {
            this.lockey = lockey;
            this.type = type;
        }

        void cleanup() {
            if (type == LockeyHolderType.WRITE) {
                lockey.wUnlock();
            } else {
                lockey.rUnlock();
            }
        }

        @Override
        public int compareTo(LockeyHolder o) {
            int c = lockey.compareTo(o.lockey);
            return c != 0 ? c : type.ordinal() - o.type.ordinal();
        }
    }
}
